package com.xrui.hbase.conversion;

import com.google.common.base.Preconditions;
import com.google.protobuf.AbstractMessage;
import com.google.protobuf.ByteString;
import com.google.protobuf.Descriptors;
import com.googlecode.protobuf.format.JsonFormat;
import com.xrui.hbase.Constants;
import com.xrui.hbase.HBaseCell;
import com.xrui.hbase.model.RecordFields;

import java.io.IOException;

import static com.googlecode.protobuf.format.util.TextUtils.escapeBytes;

/**
 * Converts an HBase cell encoded as a Protocol Buffer.
 */
public class ProtobufCellConverter implements CellConverter {
    /* The name of this provider */
    public static final String PROVIDER_NAME = "ProtobufConverter";


    private Object getFieldValue(Descriptors.FieldDescriptor field, Object value) throws IOException {
        Object fieldValue = null;
        switch (field.getType()) {
            case INT32:
            case INT64:
            case SINT32:
            case SINT64:
            case SFIXED32:
            case SFIXED64:
            case FLOAT:
            case DOUBLE:
            case BOOL:
            case UINT32:
            case FIXED32:
            case UINT64:
            case FIXED64:
            case STRING:
                fieldValue = value;
                break;
            case BYTES:
                fieldValue = escapeBytes((ByteString) value);
                break;
            case ENUM:
                fieldValue = ((Descriptors.EnumValueDescriptor) value).getName();
                break;
            case MESSAGE:
            case GROUP:
                ExtendedJsonFormat jsonFormat = new ExtendedJsonFormat();
                fieldValue = "{" + jsonFormat.getFieldValue(field, value) + "}";
                break;
        }
        return fieldValue;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public RecordFields convert(HBaseCell cell) throws IOException {
        AbstractMessage message = (AbstractMessage) cell.getData();
        Preconditions.checkArgument(
            message.getUnknownFields().asMap().size() == 0,
            "Expect none unknown fields but got :" + message.getUnknownFields()
        );
        RecordFields.Builder builder = RecordFields.Builder.newBuilder();
        builder.putField(Constants.ROW_KEY_FIELD_NAME, cell.getEntityId().toString());
        builder.putField(Constants.COLUMN_NAME_FIELD_NAME, cell.getColumn().toString());
        builder.putField(Constants.TIMESTAMP_FIELD_NAME, cell.getTimestamp());
        for (Descriptors.FieldDescriptor fieldDescriptor : message.getDescriptorForType().getFields()) {
            String name = fieldDescriptor.getName();
            Object value = getFieldValue(fieldDescriptor, message.getField(fieldDescriptor));
            builder.putField(name, value);
        }

        return builder.build();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public String getName() {
        return PROVIDER_NAME;
    }

    /**
     * An inner class for extend the JsonFormat.
     */
    protected class ExtendedJsonFormat extends JsonFormat {
        String getFieldValue(Descriptors.FieldDescriptor field, Object value) throws IOException {
            Appendable appender = new StringBuilder();
            JsonGenerator jsonGenerator = new JsonGenerator(appender);
            printField(field, value, jsonGenerator);
            return appender.toString();
        }
    }
}
